package com.platform.apex.util;

import com.google.common.base.Predicate;
import com.google.common.base.Throwables;

import org.apache.commons.dbcp2.BasicDataSource;
import org.skife.jdbi.v2.Batch;
import org.skife.jdbi.v2.DBI;
import org.skife.jdbi.v2.Handle;
import org.skife.jdbi.v2.TransactionCallback;
import org.skife.jdbi.v2.TransactionStatus;
import org.skife.jdbi.v2.exceptions.DBIException;
import org.skife.jdbi.v2.exceptions.UnableToExecuteStatementException;
import org.skife.jdbi.v2.exceptions.UnableToObtainConnectionException;
import org.skife.jdbi.v2.tweak.HandleCallback;
import org.skife.jdbi.v2.util.ByteArrayMapper;
import org.skife.jdbi.v2.util.IntegerMapper;

import java.sql.SQLException;
import java.sql.SQLRecoverableException;
import java.sql.SQLTransientException;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Callable;

public abstract class SQLMetadataConnector implements MetadataStorageConnector {

	private static final Logger log = new Logger(SQLMetadataConnector.class);

	public static final int DEFAULT_MAX_TRIES = 10;

	private final MetadataStorageConnectorConfig config;
	private final MetadataStorageTablesConfig tablesConfig;
	private final Predicate<Throwable> shouldRetry;

	public SQLMetadataConnector(MetadataStorageConnectorConfig config, MetadataStorageTablesConfig tablesConfig) {
		this.config = config;
		this.tablesConfig = tablesConfig;
		this.shouldRetry = new Predicate<Throwable>() {
			@Override
			public boolean apply(Throwable e) {
				return isTransientException(e);
			}
		};
	}

	/**
	 * Returns the value that should be passed to statement.setFetchSize to
	 * ensure results are streamed back from the database instead of fetching
	 * the entire result set in memory.
	 *
	 * @return optimal fetch size to stream results back
	 */
	protected abstract int getStreamingFetchSize();

	public String getValidationQuery() {
		return "SELECT 1";
	}

	public abstract boolean tableExists(Handle handle, final String tableName);

	public <T> T retryWithHandle(final HandleCallback<T> callback, final Predicate<Throwable> myShouldRetry) {
		final Callable<T> call = new Callable<T>() {
			@Override
			public T call() throws Exception {
				return getDBI().withHandle(callback);
			}
		};
		try {
			return RetryUtils.retry(call, myShouldRetry, DEFAULT_MAX_TRIES);
		} catch (Exception e) {
			throw Throwables.propagate(e);
		}
	}

	public <T> T retryWithHandle(final HandleCallback<T> callback) {
		return retryWithHandle(callback, shouldRetry);
	}

	public <T> T retryTransaction(final TransactionCallback<T> callback, final int quietTries, final int maxTries) {
		final Callable<T> call = new Callable<T>() {
			@Override
			public T call() throws Exception {
				return getDBI().inTransaction(callback);
			}
		};
		try {
			return RetryUtils.retry(call, shouldRetry, quietTries, maxTries);
		} catch (Exception e) {
			throw Throwables.propagate(e);
		}
	}

	public final boolean isTransientException(Throwable e) {
		return e != null && (e instanceof SQLTransientException || e instanceof SQLRecoverableException
				|| e instanceof UnableToObtainConnectionException || e instanceof UnableToExecuteStatementException
				|| connectorIsTransientException(e) || (e instanceof SQLException && isTransientException(e.getCause()))
				|| (e instanceof DBIException && isTransientException(e.getCause())));
	}

	protected boolean connectorIsTransientException(Throwable e) {
		return false;
	}

	public void createTable(final String tableName, final Iterable<String> createTableSql,
			final Iterable<String> initDataSql) {
		try {
			retryWithHandle(new HandleCallback<Void>() {
				@Override
				public Void withHandle(Handle handle) throws Exception {
					if (!tableExists(handle, tableName)) {
						log.info("Creating table[%s]", tableName);
						final Batch batch = handle.createBatch();
						for (String s : createTableSql) {
							batch.add(s);
						}

						if (null != initDataSql) {
							for (String s : initDataSql) {
								batch.add(s);
							}
						}

						batch.execute();
					} else {
						log.info("Table[%s] already exists", tableName);
					}
					return null;
				}
			});
		} catch (Exception e) {
			log.warn(e, "Exception creating table");
		}
	}

	public void createUsedTables() {
		if (config.isCreateTables()) {
			Map<String, TableDDL> usedTableDDLMap = tablesConfig.getUsedTableDDL();
			Iterator<String> iter = usedTableDDLMap.keySet().iterator();
			while (iter.hasNext()) {
				String tableName = iter.next();
				TableDDL tableDDL = usedTableDDLMap.get(tableName);
				createTable(tableDDL.tableName(), tableDDL.tableDDL(), tableDDL.initData());

			}
		}
	}

	public StringBuilder ddlDescription(){
		StringBuilder descBuilder = new StringBuilder();
		Map<String, TableDDL> usedTableDDLMap = tablesConfig.getUsedTableDDL();
		Iterator<String> iter = usedTableDDLMap.keySet().iterator();
		while (iter.hasNext()) {
			String tableName = iter.next();
			TableDDL tableDDL = usedTableDDLMap.get(tableName);
			descBuilder.append("================= ");
			descBuilder.append(tableDDL.tableName());
			descBuilder.append(" =================\n");
			descBuilder.append(tableDDL.description());
			descBuilder.append("\n");
		}
		return descBuilder;
	}
	
	@Override
	public Void insertOrUpdate(final String tableName, final String keyColumn, final String valueColumn,
			final String key, final byte[] value) throws Exception {
		return getDBI().inTransaction(new TransactionCallback<Void>() {
			@Override
			public Void inTransaction(Handle handle, TransactionStatus transactionStatus) throws Exception {
				int count = handle
						.createQuery(String.format("SELECT COUNT(*) FROM %1$s WHERE %2$s = :key", tableName, keyColumn))
						.bind("key", key).map(IntegerMapper.FIRST).first();
				if (count == 0) {
					handle.createStatement(String.format("INSERT INTO %1$s (%2$s, %3$s) VALUES (:key, :value)",
							tableName, keyColumn, valueColumn)).bind("key", key).bind("value", value).execute();
				} else {
					handle.createStatement(String.format("UPDATE %1$s SET %3$s=:value WHERE %2$s=:key", tableName,
							keyColumn, valueColumn)).bind("key", key).bind("value", value).execute();
				}
				return null;
			}
		});
	}

	public abstract DBI getDBI();

	@Override
	public byte[] lookup(final String tableName, final String keyColumn, final String valueColumn, final String key) {
		return getDBI().withHandle(new HandleCallback<byte[]>() {
			@Override
			public byte[] withHandle(Handle handle) throws Exception {
				return lookupWithHandle(handle, tableName, keyColumn, valueColumn, key);
			}
		});
	}

	public byte[] lookupWithHandle(final Handle handle, final String tableName, final String keyColumn,
			final String valueColumn, final String key) {
		final String selectStatement = String.format("SELECT %s FROM %s WHERE %s = :key", valueColumn, tableName,
				keyColumn);

		List<byte[]> matched = handle.createQuery(selectStatement).bind("key", key).map(ByteArrayMapper.FIRST).list();

		if (matched.isEmpty()) {
			return null;
		}

		if (matched.size() > 1) {
			throw new ISE("Error! More than one matching entry[%d] found for [%s]?!", matched.size(), key);
		}

		return matched.get(0);
	}

	public MetadataStorageConnectorConfig getConfig() {
		return config;
	}

	protected BasicDataSource getDatasource() {

		MetadataStorageConnectorConfig connectorConfig = getConfig();

		BasicDataSource dataSource = new BasicDataSource();
		dataSource.setUsername(connectorConfig.getUser());
		dataSource.setPassword(connectorConfig.getPassword());
		String uri = connectorConfig.getConnectURI();
		dataSource.setUrl(uri);

		dataSource.setValidationQuery(getValidationQuery());
		dataSource.setTestOnBorrow(true);

		return dataSource;
	}
}
